package com.gxlevi.hdfs.api;

import org.apache.commons.io.IOUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;
import org.junit.Test;

import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.net.URI;
import java.net.URISyntaxException;
import java.net.URL;

public class HDFSApiDemo {
    @Test
    public void urlHDFS() throws IOException {
        //1.注册url
        URL.setURLStreamHandlerFactory(new FsUrlStreamHandlerFactory());
        //2.获取hdfs文件的输入流
        InputStream inputStream = new URL("hdfs://node01:8020/a.txt").openStream();
        //3.获取本地文件的输出流
        FileOutputStream outputStream = new FileOutputStream(new File("D:\\hello2.txt"));
        //4.实现文件的拷贝
        IOUtils.copy(inputStream, outputStream);
        //5.关流
        IOUtils.closeQuietly(inputStream);
        IOUtils.closeQuietly(outputStream);
    }

    /**
     * 获取FileSystem:方式1
     *
     * @throws IOException
     */
    @Test
    public void getFileSystem1() throws IOException {
        //1.创建Configuration对象
        Configuration configuration = new Configuration();
        //2.设置文件系统的类型
        configuration.set("fs.defaultFS", "hdfs://node01:8020");
        //3.获取指定的文件系统
        FileSystem fileSystem = FileSystem.get(configuration);
        //4.输出
        System.out.println(fileSystem);
    }

    /**
     * 获取FileSystem:方式2
     *
     * @throws IOException
     */
    @Test
    public void getFileSystem2() throws IOException, URISyntaxException {
        FileSystem fileSystem = FileSystem.get(new URI("hdfs://node01:8020"), new Configuration());
        System.out.println(fileSystem);
    }

    /**
     * 获取FileSystem:方式3
     *
     * @throws IOException
     */
    @Test
    public void getFileSystem3() throws IOException {
        //1.创建Configuration对象
        Configuration configuration = new Configuration();
        //2.设置文件系统的类型
        configuration.set("fs.defaultFS", "hdfs://node01:8020");
        //3.获取指定的文件系统
        FileSystem fileSystem = FileSystem.newInstance(configuration);
        //4.输出
        System.out.println(fileSystem);
    }

    /**
     * 获取FileSystem:方式4
     *
     * @throws IOException
     */
    @Test
    public void getFileSystem4() throws IOException, URISyntaxException {
        FileSystem fileSystem = FileSystem.newInstance(new URI("hdfs://node01:8020"), new Configuration());
        System.out.println(fileSystem);
    }

    /**
     * 实现hdfs文件的遍历
     */
    @Test
    public void listFiles() throws URISyntaxException, IOException {
        FileSystem fileSystem = FileSystem.get(new URI("hdfs://node01:8020"), new Configuration());
        //调用listFiles获取/目录下的所有文件信息
        RemoteIterator<LocatedFileStatus> locatedFileStatusRemoteIterator = fileSystem.listFiles(new Path("/"), true);//true->递归获取
        while (locatedFileStatusRemoteIterator.hasNext()) {
            LocatedFileStatus fileStatus = locatedFileStatusRemoteIterator.next();
            //获取文件的绝对路径:hdfs://node01:8020/xxx
            Path path = fileStatus.getPath();
            System.out.println(path + "-------" + path.getName());
            //获取文件的block信息
            BlockLocation[] blockLocations = fileStatus.getBlockLocations();
            System.out.println("block数" + blockLocations.length);
        }
    }

    /**
     * hdfs创建文件夹
     */
    @Test
    public void mkdirsTest() throws URISyntaxException, IOException {
        FileSystem fileSystem = FileSystem.get(new URI("hdfs://node01:8020"), new Configuration());
        //2.创建文件夹
//        boolean b1 = fileSystem.mkdirs(new Path("/aaa10/abc/ccc/ddd"));
        //3.创建文件
        fileSystem.create(new Path("/bbb10/dfds/a.txt"));
//        System.out.println(b1);
        fileSystem.close();
    }

    /**
     * 实现文件的下载
     * 方式一
     */
    @Test
    public void downLoadFile() throws URISyntaxException, IOException {

        FileSystem fileSystem = FileSystem.get(new URI("hdfs://node01:8020"), new Configuration());
        //获取hdfs的输入流
        FSDataInputStream inputStream = fileSystem.open(new Path("/a.txt"));
        //获取本地路径的输出流
        FileOutputStream outputStream = new FileOutputStream("D:\\a.txt");
        //文件的copy
        IOUtils.copy(inputStream, outputStream);
        //关闭流
        IOUtils.closeQuietly(inputStream);
        IOUtils.closeQuietly(outputStream);
    }

    /**
     * 实现文件的下载
     * 方式二
     */
    @Test
    public void downLoadFile2() throws URISyntaxException, IOException, InterruptedException {
//        FileSystem fileSystem = FileSystem.get(new URI("hdfs://node01:8020"), new Configuration());
        //伪装用户
        FileSystem fileSystem = FileSystem.get(new URI("hdfs://node01:8020"), new Configuration(),"root");
        //调用方法实现文件的下载
        fileSystem.copyToLocalFile(new Path("/a.txt"), new Path("D:\\a4.txt"));
        //关闭fileSystem
        fileSystem.close();
    }

    /**
     * 实现文件的上传
     */
    @Test
    public void uploadFile() throws IOException, URISyntaxException {
        FileSystem fileSystem = FileSystem.get(new URI("hdfs://node01:8020"), new Configuration());
        fileSystem.copyFromLocalFile(new Path("D:\\a.txt"),new Path("/aaa10"));
        fileSystem.close();
    }

    /**
     * 小文件的合并
     */
    @Test
    public void mergeFile() throws URISyntaxException, IOException, InterruptedException {
        FileSystem fileSystem = FileSystem.get(new URI("hdfs://node01:8020"), new Configuration(),"root");
        //获取hdfs大文件的输出流
        FSDataOutputStream outputStream = fileSystem.create(new Path("/bigtxt.txt"));
        //获取一个本地文件系统
        LocalFileSystem localFileSystem = FileSystem.getLocal(new Configuration());
        //获取本地文件夹下所有文件的详情
        FileStatus[] fileStatuses = localFileSystem.listStatus(new Path("D:\\input"));
        //遍历每一个文件,获取每个文件的输入流
        for (FileStatus fileStatus : fileStatuses) {
            FSDataInputStream inputStream = localFileSystem.open(fileStatus.getPath());
            //将小文件的数据复制到大文件
            IOUtils.copy(inputStream,outputStream);
            IOUtils.closeQuietly(inputStream);
        }
        //关闭流
        IOUtils.closeQuietly(outputStream);
        localFileSystem.close();
        fileSystem.close();
    }

}
